package com.dtkavin.akka

import java.util.UUID

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

/**
  * Created by Administrator on 2016/4/9 0009.
  */
class Worker(val host: String, val port: Int) extends Actor {
  val workerId: String = UUID.randomUUID().toString;
  var master: ActorSelection = _

  override def preStart(): Unit = {
    println(s"worker 1 [$host: $port]  starting ...")

    master = context.actorSelection("akka.tcp://MasterSystem@127.0.0.1:9900/user/Master")

    master ! WorkderRegister(workerId, host, port, 256, 2)
    println(workerId)
  }

  override def receive: Receive = {
    case RegisterBack(masterId) => {
      import scala.concurrent.duration._
      import scala.concurrent.ExecutionContext.Implicits.global
      context.system.scheduler.schedule(0 millis, 2000 millis, self, WorkerHearBeat)
    }
    case WorkerHearBeat => master ! HeartBeat(workerId, System.currentTimeMillis())
    case _ => println("worker received mismatched msg")
  }
}

object Worker {
  def apply(host: String, port: Int) = new Worker(host, port)

  def main(args: Array[String]) {
    val host = "127.0.0.1"
    val port = 9911

    val confStr =
      s"""
         akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |                      |akka.remote.netty.tcp.hostname = "$host"
         |                      |akka.remote.netty.tcp.port = "$port"""" stripMargin

    val conf = ConfigFactory.parseString(confStr)
    val workerSystem1 = ActorSystem("Worker1", conf)
    workerSystem1.actorOf(Props(Worker(host, port)), "Worker1")
    workerSystem1.awaitTermination()
  }
}
